package br.ufpe.cin.dsoa.qos.epcenter.impl;

import java.util.Dictionary;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;
import org.osgi.service.event.EventConstants;
import org.osgi.service.event.EventHandler;

import br.ufpe.cin.dsoa.qos.epcenter.service.EventConsumer;
import br.ufpe.cin.dsoa.qos.epcenter.service.EventProcessingCenter;
import br.ufpe.cin.log.generator.service.LogGenerator;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;

public class EventProcessingCenterImpl implements EventProcessingCenter {

	private final EPServiceProvider epServiceProvider;
	private final BundleContext bundleContext;
	private final Map<String, EventNotifier> notifierMap;
	private LogGenerator logGenerator;

	public EventProcessingCenterImpl(BundleContext context) {
		this.bundleContext = context;
		this.notifierMap = new HashMap<String, EventNotifier>();
		this.epServiceProvider = EPServiceProviderManager.getProvider(
				"EngineInstance", new Configuration());
		
		ServiceReference reference = context.getServiceReference(LogGenerator.class.getName());
		if (reference != null) {
			this.logGenerator = (LogGenerator) context.getService(reference);
		}
	}

	@Override
	public void publishEvent(Object event) {
		
/*		if(event.getClass().getSimpleName().equalsIgnoreCase("RequestEvent") == true){
			DateFormat df = new SimpleDateFormat("dd/MM/yyyy hh:mm:ss.SSS");
			StringBuilder builder = new StringBuilder(1000);
			builder.append("RequestEvent").append(" - ");
			builder.append(df.format(new Date(System.currentTimeMillis())));
			logGenerator.log(Level.INFO, builder.toString());
		} */
		
		this.epServiceProvider.getEPRuntime().sendEvent(event);
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void publishEvent(Map event, String eventName) {
		Map<String, Object> propriedades = event;
		this.epServiceProvider.getEPRuntime().sendEvent(event, eventName);
	}

	@Override
	public void defineStatement(String name, String statement, List<String> userObject) {
		this.epServiceProvider.getEPAdministrator().createEPL(statement, name, userObject);
	}
	
	public void defineStatement(String name, String statement) {
		this.epServiceProvider.getEPAdministrator().createEPL(statement, name);
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void defineEvent(Class event) {
		this.epServiceProvider.getEPAdministrator().getConfiguration().addEventType(event);
		Dictionary<Object, Object> propriedades = new Hashtable<Object, Object>();
		String[] topics = new String[] { event.getSimpleName() };
		propriedades.put(EventConstants.EVENT_TOPIC, topics);
		this.bundleContext.registerService(EventHandler.class.getName(),
				new EventHandlerImpl(), propriedades);
	}

	@Override
	public void subscribe(String nameStatement, EventConsumer eventConsumer) {
		EventNotifier notifier = notifierMap.get(nameStatement);
		if (null == notifier) {
			notifier = new EventNotifier(epServiceProvider);
			this.notifierMap.put(nameStatement, notifier);
		}
		notifier.addEventConsumer(eventConsumer);
		this.epServiceProvider.getEPAdministrator().getStatement(nameStatement).addListener(notifier);
	}

	public void unsubscribe(String nameStatement, EventConsumer eventConsumer) {
		EventNotifier notifier = this.notifierMap.get(nameStatement);
		notifier.removeEventConsumer(eventConsumer);
		if (!notifier.hasEventConsumers()) {
			this.epServiceProvider.getEPAdministrator()
					.getStatement(nameStatement).destroy();
		}
	}

	@Override
	public void destroyStatement(String statementName) {
		if (this.epServiceProvider.getEPAdministrator().getStatement(statementName) != null) {
			this.epServiceProvider.getEPAdministrator().getStatement(statementName).destroy();
			this.notifierMap.remove(statementName);
		}

	}

	class EventHandlerImpl implements EventHandler {
		private int poolSize = 30;
		private int maxPoolSize = 50;
		private long keepAliveTime = 10;
		final ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(
	            100);
		
		private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(poolSize, maxPoolSize,
                keepAliveTime, TimeUnit.SECONDS, queue);
		
		@Override
		public void handleEvent(org.osgi.service.event.Event event) {
			final Object eventReceived = event.getProperty(event.getTopic());
			Runnable publisher = new Runnable() {
				@Override
				public void run() {
					publishEvent(eventReceived);
				}
			};
			threadPool.execute(publisher);
		}

	}

}
